package com.ihr360.binlog.component;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.ihr360.binlog.daemon.BinlogReceiver;
import com.ihr360.binlog.dao.CanalConfigRepository;
import com.ihr360.binlog.dto.CanalConfig;
import com.ihr360.binlog.service.BinlogEventSourceService;
import com.ihr360.binlog.service.IProcessErrorCallback;
import com.ihr360.binlog.service.IProcessEventEntry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class BinlogReceiverStarter {
    final Logger logger = LoggerFactory.getLogger(BinlogReceiverStarter.class.getName());
    @Autowired
    private CanalConfigRepository canalConfigRepository;
    @Autowired
    private BinlogEventSourceService binlogEventSourceService;

    private Map<Long, BinlogReceiver> binlogReceiverMap;

    private Boolean onError = false;

    private String errorMessage;

    public Boolean getOnError() {
        return onError;
    }

    public String getErrorMessage() {
        return errorMessage;
    }


    public void startAll() {
        synchronized (this) {
            binlogReceiverMap = new ConcurrentHashMap<>();
            Iterable<CanalConfig> canalConfigs = canalConfigRepository.findAll();
            Iterator<CanalConfig> iterator = canalConfigs.iterator();
            while (iterator.hasNext()) {
                CanalConfig c = iterator.next();
                if (c.getIsEnabled()!=null && c.getIsEnabled()==true){
                    BinlogReceiver receiver = new BinlogReceiver(c, iprocessEvent, iProcessErrorCallback);
                    receiver.start();
                    c.setIsUp(true);
                    c.setError(false);
                    binlogReceiverMap.put(c.getId(), receiver);
                }
            }
            canalConfigRepository.save(canalConfigs);
            onError = false;
        }
    }

    public void stopAll() {
        synchronized (this) {
            for (Map.Entry<Long, BinlogReceiver> entry : binlogReceiverMap.entrySet()) {
                Long id = entry.getKey();
                BinlogReceiver receiver = entry.getValue();
                receiver.stop();
                CanalConfig config = canalConfigRepository.findOne(id);
                if (config.getIsUp() == null || config.getIsUp().equals(Boolean.TRUE)) {
                    config.setIsUp(false);
                    canalConfigRepository.save(config);
                }
            }
            binlogReceiverMap.clear();
        }
    }

    @Scheduled(fixedDelay = 30000)
    public void refresh() {
        synchronized (this) {
            if (binlogReceiverMap==null) {
                binlogReceiverMap = new ConcurrentHashMap<>();
            }
            Iterable<CanalConfig> canalConfigs = canalConfigRepository.findAll();
            Iterator<CanalConfig> iterator = canalConfigs.iterator();
            while (iterator.hasNext()) {
                CanalConfig c = iterator.next();
                BinlogReceiver receiver = binlogReceiverMap.get(c.getId());
                if (receiver == null || !receiver.isOn()) {
                    receiver = new BinlogReceiver(c, iprocessEvent, iProcessErrorCallback);
                    receiver.start();
                    c.setIsUp(true);
                    c.setError(false);
                    binlogReceiverMap.put(c.getId(), receiver);
                }
            }
            onError = false;
        }
    }

    IProcessEventEntry iprocessEvent = new IProcessEventEntry() {
        @Override
        public boolean processEventEntrys(List<CanalEntry.Entry> entrys, CanalConfig config) {
            return binlogEventSourceService.processEventEntrys(entrys,config);
        }
    };

    IProcessErrorCallback iProcessErrorCallback = new IProcessErrorCallback() {
        @Override
        public void onError(CanalConfig config, Exception e) {
            logger.error("configId:" + config.getId() + " " + e.getMessage(), e.getCause());
            errorMessage = e.getMessage();
            onError = true;
            config.setError(true);
            config.setIsUp(false);
            canalConfigRepository.save(config);
            binlogReceiverMap.remove(config.getId());
        }
    };
}